package fox.framework.rpc.server.dispatcher;

import fox.framework.rpc.common.ExceptionUtils;
import fox.framework.rpc.common.StringUtils;
import fox.framework.rpc.converter.RpcMessageConverter;
import fox.framework.rpc.converter.RpcMessageConverters;
import fox.framework.rpc.exception.RpcConverterException;
import fox.framework.rpc.exception.RpcRuntimeException;
import fox.framework.rpc.message.RpcHeader;
import fox.framework.rpc.message.RpcMessageStatus;
import fox.framework.rpc.server.dispatcher.invoke.ProvideInvoker;
import fox.framework.rpc.server.dispatcher.invoke.ProvideInvokerChain;
import fox.framework.rpc.server.dispatcher.invoke.ReflectionProvideInvokerChain;
import fox.framework.rpc.server.dispatcher.router.DataBinder;
import fox.framework.rpc.server.dispatcher.router.RouteMapping;
import fox.framework.rpc.server.dispatcher.router.RpcRouter;
import fox.framework.rpc.server.message.request.RpcAttribute;
import fox.framework.rpc.server.message.request.RpcServerRequest;
import fox.framework.rpc.server.message.response.RpcServerResponse;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * @author cuichao
 * @Description: AbstractRpcRequestDispatcher
 * @Date: create in 2021/1/12 16:01
 */
public abstract class AbstractRpcRequestDispatcher implements RpcRequestDispatcher {

    private RpcRouter router;
    private List<ProvideInvoker> chains;
    private RpcMessageConverters converters;
    private ConcurrentMap<String, RpcMessageConverter> converterCache = new ConcurrentHashMap<>(16);


    public AbstractRpcRequestDispatcher(RpcMessageConverters converters, RpcRouter router, List<ProvideInvoker> chains) {
        this.router = router;
        this.chains = chains;
        this.converters = converters;
    }

    @Override
    public void dispatch(RpcServerRequest request, RpcServerResponse response) {
        try {
            doDispatch(request, response);
        } catch (Throwable throwable) {
            processExceptionHandler(response, throwable);
        }
    }

    private void doDispatch(RpcServerRequest request, RpcServerResponse response) throws Throwable {
        //check request-id
        checkRequestId(request, response);
        //get mapping
        RouteMapping mapping = getMapping(request);
        //set attr converter
        matchConverter(request, response);
        //set attr parameterTypes
        getDataBinders(request, mapping.getDataBinders());
        //do invoker chains
        ProvideInvokerChain chain = new ReflectionProvideInvokerChain(chains, mapping);
        chain.doChain(request, response);
    }


    @Override
    public void processExceptionHandler(RpcServerResponse response, Throwable throwable) {
        RpcMessageConverter converter = matchConverter(response.getProtocol(),RpcMessageStatus.COMMON_EXCEPTION_TYPE);
        response.setStatus(RpcMessageStatus.COMMON_EXCEPTION_CODE);
        response.setContentType(RpcMessageStatus.COMMON_EXCEPTION_TYPE);
        response.setAttribute(RpcAttribute.CONVERTER_NAME, converter);
        String msg = throwable.getMessage();
        response.setEntityBody(msg);
    }


    private void checkRequestId(RpcServerRequest request, RpcServerResponse response) {
        String requestId = request.getHeader(RpcHeader.REQUEST_ID);
        if (!StringUtils.isEmpty(requestId)) {
            response.setHeader(RpcHeader.REQUEST_ID, requestId);
        }
    }

    private RouteMapping getMapping(RpcServerRequest request) {
        String path = request.getPath();
        RouteMapping mapping = router.getMapping(path);
        if (mapping == null) {
            try {
                mapping = processNoRouteMappingHandler(router, request);
            } catch (RpcRuntimeException e) {
                throw e;
            } catch (Exception e) {
                throw ExceptionUtils.RpcRuntimeExceptionWrap(e);
            }
        }
        if (mapping == null) {
            throw new RpcRuntimeException("the request url is not mapping at rpc server");
        }
        return mapping;
    }


    private void matchConverter(RpcServerRequest request, RpcServerResponse response) {
        String contentType = request.getContentType();
        RpcMessageConverter converter = matchConverter(request.getProtocol(),contentType);
        request.setAttribute(RpcAttribute.CONVERTER_NAME, converter);
        response.setAttribute(RpcAttribute.CONVERTER_NAME, converter);
        response.setContentType(contentType);
    }


    private RpcMessageConverter matchConverter(String protocol,String contentType) {
        String key = protocol+":"+contentType;
        RpcMessageConverter converter = converterCache.getOrDefault(key, null);
        if (converter == null) {
            //find out to converters
            converter = converters.getSupportConverter(protocol,contentType);
        }
        if (converter == null) {
            String msg = String.format("the request is not encoder or decoder by content-type = %s ", contentType);
            throw new RpcConverterException(msg);
        }
        converterCache.put(key, converter);
        return converter;
    }


    private void getDataBinders(RpcServerRequest request, DataBinder[] dataBinders) {
        request.setAttribute(RpcAttribute.DATA_BINDER, dataBinders);
    }


    /**
     * 处理没有映射的Request
     *
     * @param request
     * @param router
     * @return
     * @throws RpcRuntimeException
     */
    protected abstract RouteMapping processNoRouteMappingHandler(RpcRouter router, RpcServerRequest request) throws RpcRuntimeException;


}
